package rabbitmq.ack.direct;

import com.alibaba.fastjson.JSONArray;
import com.google.common.collect.Lists;
import com.rabbitmq.client.Channel;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.RabbitHandler;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import java.util.ArrayList;
import java.util.LinkedList;
import java.util.Queue;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

@Component
@RabbitListener(queues = "ackDirectQueue")
public class DirectGetServiceB {

    @Autowired
    private MyExecutorService myExecutorService;


//    @RabbitHandler
//    public void process(String context, Channel channel, Message message){
//        System.out.println("HelloReceiver收到  : " + context );
//        try {
//            //告诉服务器收到这条消息 已经被我消费了 可以在队列删掉 这样以后就不会再发了 否则消息服务器以为这条消息没处理掉 后续还会在发
//            channel.basicAck(message.getMessageProperties().getDeliveryTag(),false);
//            System.out.println("receiver success");
//        } catch (Exception e) {
//            e.printStackTrace();
//            System.out.println("receiver fail");
//        }
//    }

    @RabbitHandler
    public void process(String context, Channel channel, Message message) {
        System.out.println("HelloReceiverB收到  : " + context);
        try {
            ArrayList<String> robot = Lists.newArrayList("A", "B", "C", "D");

            Queue<String> queue = new ArrayBlockingQueue(4);
            queue.add("A");
            queue.add("B");
            queue.add("C");
            queue.add("D");

            LinkedList<User> users1 = new LinkedList<>(JSONArray.parseArray(context, User.class));
            ExecutorService executor = new ThreadPoolExecutor(4, 4, 5, TimeUnit.MILLISECONDS,
                    new ArrayBlockingQueue<Runnable>(12), new ThreadPoolExecutor.AbortPolicy());
            while (true) {
                if (users1.size() == 0) break;
                User user = users1.removeFirst();
                String poll = queue.poll();
                queue.add(poll);
                executor.submit(new Runnable() {
                    @Override
                    public void run() {
                        myExecutorService.executor(user, poll);
                    }
                });
            }
            executor.shutdown();

            //告诉服务器收到这条消息 已经被我消费了 可以在队列删掉 这样以后就不会再发了 否则消息服务器以为这条消息没处理掉 后续还会在发
            channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
            System.out.println("receiverB success");
        } catch (Exception e) {
            e.printStackTrace();
            System.out.println("receiverB fail");
        }
    }

//    @RabbitHandler
//    public void process2(ArrayList<User> context, Channel channel, Message message) {
//        System.out.println("HelloReceiver收到  : " + context);
//        try {
//            //告诉服务器收到这条消息 已经被我消费了 可以在队列删掉 这样以后就不会再发了 否则消息服务器以为这条消息没处理掉 后续还会在发
//            channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
//            System.out.println("receiver success");
//        } catch (Exception e) {
//            e.printStackTrace();
//            System.out.println("receiver fail");
//        }
//    }
}
